KAFKA-19144 Move DelayedProduce to server module #19793

johnny94 · 2025-05-23T09:16:05Z

This PR moves DelayedProduce to the server module. One notable change is that the type of the responseCallback parameter in ReplicaManager#appendRecords() has been changed to a Java Map. Other related type changes have been made accordingly.

FrankYang0529

Thanks for the patch. Leave some minor comments.

FrankYang0529 · 2025-05-24T04:02:33Z

core/src/main/scala/kafka/server/ReplicaManager.scala

      val produceResponseStatus = initialProduceStatus.map { case (k, status) => k -> status.responseStatus }
-      responseCallback(produceResponseStatus)
+      responseCallback(produceResponseStatus.asJava)


We use initialProduceStatus to build a Scala map and transfer to Java. Probably, we can build a Java map directly, so we can avoid asJava here.

val produceResponseStatus = new util.HashMap[TopicIdPartition, PartitionResponse]() initialProduceStatus.foreach { case (k, status) => produceResponseStatus.put(k, status.responseStatus()) } responseCallback(produceResponseStatus)

FrankYang0529 · 2025-05-24T04:03:04Z

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -1006,7 +1026,7 @@ class ReplicaManager(val config: KafkaConfig,
        LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.logStartOffset
      )
    }
-    responseCallback(responseStatus)
+    responseCallback(responseStatus.asJava)


ditto

val responseStatus = new util.HashMap[TopicIdPartition, PartitionResponse]() entries.foreach { case (topicIdPartition, _) => responseStatus.put(topicIdPartition, new PartitionResponse( Errors.INVALID_REQUIRED_ACKS, LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.firstOffset, RecordBatch.NO_TIMESTAMP, LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.logStartOffset )) } responseCallback(responseStatus)

chia7712 · 2025-05-25T06:49:41Z

I will review this PR after #19798 gets merged.

chia7712 · 2025-05-26T17:06:44Z

@johnny94 please fix the conflicts

frankvicky

@johnny94 Thanks for the patch!
I have a few comments. PTAL

frankvicky · 2025-05-27T02:53:57Z

core/src/main/scala/kafka/server/ReplicaManager.scala

+      def delegate(tp: TopicPartition, status: ProducePartitionStatus) : Unit = {
+        val (hasEnough, error) = getPartitionOrError(tp) match {
+          case Left(err) =>
+            // Case A


What does this comment mean?

I see this is relevant to the comment of DelayedProduce, but it's confusing that these comments are standalone here.
Could you write the whole meaning of these cases, or link these comments to tryComplete ?

frankvicky · 2025-05-27T02:54:13Z

core/src/main/scala/kafka/server/ReplicaManager.scala

+            partition.checkEnoughReplicasReachOffset(status.requiredOffset)
+        }
+
+        // Case B || C.1 || C.2


frankvicky · 2025-05-27T02:57:48Z

core/src/test/scala/kafka/server/LocalLeaderEndPointTest.scala

-    def appendCallback(responses: scala.collection.Map[TopicIdPartition, PartitionResponse]): Unit = {
-      val response = responses.get(partition)
-      assertTrue(response.isDefined)
+    def appendCallback(responses: java.util.Map[TopicIdPartition, PartitionResponse]): Unit = {


Given that we already have an import alias for java.util.Map, we could reuse it.

Suggested change

def appendCallback(responses: java.util.Map[TopicIdPartition, PartitionResponse]): Unit = {

def appendCallback(responses: JMap[TopicIdPartition, PartitionResponse]): Unit = {

frankvicky · 2025-05-27T03:00:06Z

core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala

@@ -1104,7 +1104,7 @@ class TransactionStateManagerTest {
    capturedAppends: mutable.Map[TopicIdPartition, mutable.Buffer[MemoryRecords]]
  ): Unit = {
    val recordsCapture: ArgumentCaptor[Map[TopicIdPartition, MemoryRecords]] = ArgumentCaptor.forClass(classOf[Map[TopicIdPartition, MemoryRecords]])
-    val callbackCapture: ArgumentCaptor[Map[TopicIdPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[Map[TopicIdPartition, PartitionResponse] => Unit])
+    val callbackCapture: ArgumentCaptor[java.util.Map[TopicIdPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[java.util.Map[TopicIdPartition, PartitionResponse] => Unit])


please import java.util

frankvicky · 2025-05-27T03:02:41Z

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

@@ -2832,7 +2832,7 @@ class KafkaApisTest extends Logging {
      any(),
      ArgumentMatchers.eq(requestLocal),
      any()
-    )).thenAnswer(_ => responseCallback.getValue.apply(Map(new TopicIdPartition(topicId,tp2) -> new PartitionResponse(Errors.NONE))))
+    )).thenAnswer(_ => responseCallback.getValue.apply(Map(new TopicIdPartition(topicId,tp2) -> new PartitionResponse(Errors.NONE)).asJava))


Suggested change

)).thenAnswer(_ => responseCallback.getValue.apply(Map(new TopicIdPartition(topicId,tp2) -> new PartitionResponse(Errors.NONE)).asJava))

)).thenAnswer(_ => responseCallback.getValue.apply(util.Map.of(new TopicIdPartition(topicId,tp2), new PartitionResponse(Errors.NONE))))

frankvicky · 2025-05-27T03:06:38Z

core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala

-      val response = responses.get(topicIdPartition)
-      assertTrue(response.isDefined)
+    def appendCallback(responses: util.Map[TopicIdPartition, PartitionResponse]): Unit = {
+      val response = java.util.Optional.ofNullable(responses.get(topicIdPartition))


We already imported java.util.Optional, so we don't need a full-qualified name.

frankvicky · 2025-05-27T03:21:37Z

server/src/main/java/org/apache/kafka/server/DelayedProduce.java

+            PARTITION_EXPIRATION_METERS.computeIfAbsent(partition,
+                    key -> METRICS_GROUP.newMeter("ExpiresPerSec", "requests", TimeUnit.SECONDS,
+                                        Map.of("topic", key.topic(), "partition", String.valueOf(key.partition()))))
+                    .mark();


style nit:

Suggested change

PARTITION_EXPIRATION_METERS.computeIfAbsent(partition,

key -> METRICS_GROUP.newMeter("ExpiresPerSec", "requests", TimeUnit.SECONDS,

Map.of("topic", key.topic(), "partition", String.valueOf(key.partition()))))

.mark();

PARTITION_EXPIRATION_METERS.computeIfAbsent(partition,

key -> METRICS_GROUP.newMeter("ExpiresPerSec",

"requests",

TimeUnit.SECONDS,

Map.of("topic", key.topic(), "partition", String.valueOf(key.partition())))

).mark();

frankvicky · 2025-05-27T03:44:06Z

server/src/main/java/org/apache/kafka/server/DelayedProduce.java

+        boolean anyPending = produceMetadata.produceStatus
+                                            .values()
+                                            .stream()
+                                            .anyMatch(ProducePartitionStatus::acksPending);
+        if (!anyPending) {
+            return forceComplete();
+        }
+
+        return false;


nit:

Suggested change

boolean anyPending = produceMetadata.produceStatus

.values()

.stream()

.anyMatch(ProducePartitionStatus::acksPending);

if (!anyPending) {

return forceComplete();

}

return false;

return produceMetadata.produceStatus.values()

.stream()

.findAny()

.map(__ -> false)

.orElseGet(this::forceComplete);

hmm, I think it keeps the idea of original implementation and easier to read. What do you think?

I'm okay with it.

m1a2st

Thanks @johnny94 for this patch, left some comments

m1a2st · 2025-05-27T07:53:57Z

core/src/main/scala/kafka/server/ReplicaManager.scala

+      val produceResponseStatus = new util.HashMap[TopicIdPartition, PartitionResponse]
+      initialProduceStatus.foreach { case (k, status) => k -> produceResponseStatus.put(k, status.responseStatus) }


Suggested change

val produceResponseStatus = new util.HashMap[TopicIdPartition, PartitionResponse]

initialProduceStatus.foreach { case (k, status) => k -> produceResponseStatus.put(k, status.responseStatus) }

val produceResponseStatus = initialProduceStatus.map { case (k, status) =>

k -> status.responseStatus

}.asJava

I think it's ok to use HashMap instead of asJava conversion.

m1a2st · 2025-05-27T07:59:40Z

core/src/test/scala/kafka/server/LocalLeaderEndPointTest.scala

+      val response = Optional.ofNullable(responses.get(partition))
+
+      assertTrue(response.isPresent)
      result.fire(response.get)


Suggested change

val response = Optional.ofNullable(responses.get(partition))

assertTrue(response.isPresent)

result.fire(response.get)

val response = responses.get(partition)

assertNotNull(response)

result.fire(response)

m1a2st · 2025-05-27T08:01:53Z

core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala

+      val response = java.util.Optional.ofNullable(responses.get(topicIdPartition))
+      assertTrue(response.isPresent)
      result.fire(response.get)


Suggested change

val response = java.util.Optional.ofNullable(responses.get(topicIdPartition))

assertTrue(response.isPresent)

result.fire(response.get)

val response = responses.get(topicIdPartition)

assertNotNull(response)

result.fire(response)

brandboat · 2025-05-27T11:29:13Z

server/src/main/java/org/apache/kafka/server/DelayedProduce.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server;


Last time, we moved DelayedDeleteRecords to org.apache.kafka.server.purgatory (see 2994e5e). I'm curious, why we chose a different location for DelayedProduce?

DL1231 · 2025-05-28T07:38:14Z

server/src/main/java/org/apache/kafka/server/purgatory/DelayedProduce.java

+            LOGGER.trace("Checking produce satisfaction for {}, current status {}", topicIdPartition, status);
+            // skip those partitions that have already been satisfied
+            if (status.acksPending) {
+                // Delegate to `ReplicaManager#maybeAddDelayedProduc`


maybeAddDelayedProduc -> maybeAddDelayedProduce

github-actions bot added triage PRs from the community core Kafka Broker labels May 23, 2025

chia7712 added the ci-approved label May 23, 2025

KAFKA-19144 Move DelayedProduce to server module

9f2951e

johnny94 force-pushed the kafka-19144_move_delayed_produce_to_server_module branch from 03e76dc to 9f2951e Compare May 23, 2025 17:04

FrankYang0529 reviewed May 24, 2025

View reviewed changes

Addressed comment

30fb277

github-actions bot removed the triage PRs from the community label May 25, 2025

Fix conflicts

458807b

frankvicky reviewed May 27, 2025

View reviewed changes

m1a2st reviewed May 27, 2025

View reviewed changes

brandboat reviewed May 27, 2025

View reviewed changes

johnny94 added 3 commits May 28, 2025 13:08

Addressed comments

c86ffc4

Addressed comments part2

b19a431

Addressed comments part3

f376a7f

DL1231 reviewed May 28, 2025

View reviewed changes

Addressed comments part4

c1409d0

	def appendCallback(responses: java.util.Map[TopicIdPartition, PartitionResponse]): Unit = {
	def appendCallback(responses: JMap[TopicIdPartition, PartitionResponse]): Unit = {

	)).thenAnswer(_ => responseCallback.getValue.apply(Map(new TopicIdPartition(topicId,tp2) -> new PartitionResponse(Errors.NONE)).asJava))
	)).thenAnswer(_ => responseCallback.getValue.apply(util.Map.of(new TopicIdPartition(topicId,tp2), new PartitionResponse(Errors.NONE))))

		val produceResponseStatus = new util.HashMap[TopicIdPartition, PartitionResponse]
		initialProduceStatus.foreach { case (k, status) => k -> produceResponseStatus.put(k, status.responseStatus) }

KAFKA-19144 Move DelayedProduce to server module #19793

Are you sure you want to change the base?

KAFKA-19144 Move DelayedProduce to server module #19793

Uh oh!

Conversation

johnny94 commented May 23, 2025

Uh oh!

FrankYang0529 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 commented May 25, 2025

Uh oh!

chia7712 commented May 26, 2025

Uh oh!

frankvicky left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

johnny94 May 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

m1a2st left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

johnny94 May 28, 2025 •

edited

Loading